#include "ThreadTask.h"

static void SyncHronize(ThreadTask_t* threadTask) {
    struct timespec tvNow;
    clock_gettime(CLOCK_MONOTONIC, &tvNow);

    if ((0 == threadTask->timeLastSync.tv_sec) && (0 == threadTask->timeLastSync.tv_nsec))
        clock_gettime(CLOCK_MONOTONIC, &(threadTask->timeLastSync));

    int64_t elapseMsec = (tvNow.tv_sec * 1000 + tvNow.tv_nsec / 1000000) - (threadTask->timeLastSync.tv_sec * 1000 + threadTask->timeLastSync.tv_nsec / 1000000);

    for (int i = 0; i < threadTask->taskNums; i++) {
        if (threadTask->tasks[i].delay < elapseMsec)
            threadTask->tasks[i].delay = 0;
        else
            threadTask->tasks[i].delay -= elapseMsec;
    }
    clock_gettime(CLOCK_MONOTONIC, &(threadTask->timeLastSync));
}

static void* ThreadTaskProcess(void* threadTask) {
    ThreadTask_t* tasks = (ThreadTask_t*)threadTask;
    ThreadTaskDelay_t func;

    for (;;) {
        // syncHronize(tasks);

        pthread_mutex_lock(&(tasks->mutex));
        while ((!tasks->taskNums) && (!tasks->shutdown)) {
            pthread_cond_wait(&(tasks->cond), &(tasks->mutex));
        }
        if ((shutdownNow == tasks->shutdown) || ((shutdownGraceFul == tasks->shutdown) && !tasks->taskNums)) {
            break;
        }
        func.function = tasks->tasks[tasks->head].function;
        func.argument = tasks->tasks[tasks->head].argument;
        func.ctx = tasks->tasks[tasks->head].ctx;

        tasks->head += 1;
        tasks->head = (tasks->head == tasks->taskSize) ? 0 : tasks->head;
        tasks->taskNums -= 1;

        pthread_mutex_unlock(&(tasks->mutex));

        func.function(func.ctx, func.argument);
        usleep(20 * 1000);
    }
    tasks->started--;

    pthread_mutex_unlock(&(tasks->mutex));
    pthread_exit(NULL);

    return NULL;
}

int32_t ThreadTaskFree(ThreadTask_t* threadTask) {
    if (!threadTask || threadTask->started > 0)
        return -1;

    if (threadTask->threads) {
        free(threadTask->threads);

        if (threadTask->tasks)
            free(threadTask->tasks);

        pthread_mutex_lock(&(threadTask->mutex));
        pthread_mutex_destroy(&(threadTask->mutex));
        pthread_cond_destroy(&(threadTask->cond));
    }

    free(threadTask);
    return 0;
}

ThreadTask_t* ThreadTaskCreate(int32_t threadSize, int32_t taskSize, int32_t flags) {
    if (threadSize <= 0 || threadSize > MAX_THREADS || taskSize <= 0 || taskSize > MAX_TASKS)
        return NULL;

    ThreadTask_t* threadTask = (ThreadTask_t*)malloc(sizeof(ThreadTask_t));

    if (!threadTask)
        return NULL;

    threadTask->threadNums = threadSize;
    threadTask->taskSize = taskSize;
    threadTask->head = threadTask->tail = threadTask->taskNums = 0;
    threadTask->shutdown = threadTask->started = 0;

    threadTask->threads = (pthread_t*)malloc(sizeof(pthread_t) * threadTask->threadNums);
    threadTask->tasks = (ThreadTaskDelay_t*)malloc(sizeof(ThreadTaskDelay_t) * threadTask->taskSize);

    if (!threadTask->threads || !threadTask->tasks)
        goto err;

    if ((pthread_mutex_init(&(threadTask->mutex), NULL)) || (pthread_cond_init(&(threadTask->cond), NULL)))
        goto err;

    for (size_t i = 0; i < threadTask->threadNums; i++) {
        if (pthread_create(&(threadTask->threads[i]), NULL, ThreadTaskProcess, (void*)threadTask)) {
            fprintf(stderr, "threadTaskProcess creat err!\r\n");
            ThreadTaskDestory(threadTask, 0);
            return NULL;
        }
        threadTask->started++;
    }
    return threadTask;

err:
    if (threadTask)
        ThreadTaskFree(threadTask);

    return NULL;
}

int32_t ThreadTaskDestory(ThreadTask_t* threadTask, int32_t flags) {
    int errno = 0;

    if (!threadTask)
        return -1;

    do {
        if (threadTask->shutdown) {
            errno = threadpoolShutdown;
            break;
        }

        threadTask->shutdown = (flags & threadpool_graceful) ? shutdownGraceFul : shutdownNow;

        if ((pthread_cond_broadcast(&(threadTask->cond))) && (pthread_mutex_unlock(&(threadTask->mutex)))) {
            errno = threadpoolLockFailure;
            break;
        }

        for (int i = 0; i < threadTask->threadNums; i++) {
            if (pthread_join(threadTask->threads[i], NULL))
                errno = threadpoolThreadFailure;
        }

    } while (0);

    if (!errno)
        ThreadTaskFree(threadTask);

    return errno;
}

int32_t ThreadTaskAdd(ThreadTask_t* threadTask, ThreadTaskFunc_t func,
                      void* arg, uint64_t delay) {

    int32_t errno = 0;
    int32_t next;
    if (!threadTask || !func)
        return -1;
    if (pthread_mutex_lock(&(threadTask->mutex)))
        return -2;
    next = threadTask->tail + 1;
    next = (next == threadTask->taskSize) ? 0 : next;
    do {
        if (threadTask->taskNums == threadTask->taskSize) {
            errno = -3;
            break;
        }
        if (threadTask->shutdown) {
            errno = -4;
            break;
        }

        threadTask->tasks[threadTask->tail].function = func;
        threadTask->tasks[threadTask->tail].ctx = threadTask;
        threadTask->tasks[threadTask->tail].argument = arg;
        threadTask->tail = next;
        threadTask->taskNums += 1;

        if (pthread_cond_signal(&(threadTask->cond))) {
            errno = -5;
            break;
        }
    } while (0);

    if (pthread_mutex_unlock(&(threadTask->mutex)))
        errno = -6;

    return errno;
}
